package com.brave.mvvm.mvvmhelper.utils

import com.brave.mvvm.mvvmhelper.http.BaseResponse
import com.brave.mvvm.mvvmhelper.http.ExceptionHandle
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.core.Observer
import io.reactivex.rxjava3.disposables.Disposable
import io.reactivex.rxjava3.schedulers.Schedulers

/**
 * ***author*** ：brave tou
 *
 * ***data***   : 2021/9/1 14:53
 *
 * ***desc***   : RxJava&RxAndroid线程调度工具
 */

/**
 * 用于文件读写，网络请求（线程池）等操作；
 * 用于IO密集型任务，如异步阻塞IO操作，这个调度器的线程池会根据需要增长；
 * 对于普通的计算任务，请使用Schedulers.computation()；
 * Schedulers.io( )默认是一个CachedThreadScheduler，很像一个有线程缓存的新线程调度器
 */
fun <T> Observable<T>.schedulersIO(): Observable<T> {
    return this.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
}

fun <T> Observable<T>.schedulersIOE(): Observable<T> {
    return this.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .onErrorResumeNext()
}

/**
 * 用于计算任务，如事件循环或和回调处理，极其占用CPU
 * 不要用于IO操作(IO操作请使用Schedulers.io())；
 * 默认线程数等于处理器的数量
 */
fun <T> Observable<T>.schedulersComputation(): Observable<T> {
    return this.subscribeOn(Schedulers.computation())
        .observeOn(AndroidSchedulers.mainThread())
}

fun <T> Observable<T>.schedulersComputationE(): Observable<T> {
    return this.subscribeOn(Schedulers.computation())
        .observeOn(AndroidSchedulers.mainThread())
        .onErrorResumeNext()
}

/**
 * 开启常规新线程
 * 为每个任务创建一个新线程
 */
fun <T> Observable<T>.schedulersNewThread(): Observable<T> {
    return this.subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
}

fun <T> Observable<T>.schedulersNewThreadE(): Observable<T> {
    return this.subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .onErrorResumeNext()
}

/**
 * 当其它排队的任务完成后
 * 在当前线程排队开始执行
 */
fun <T> Observable<T>.schedulersTrampoline(): Observable<T> {
    return this.subscribeOn(Schedulers.trampoline())
        .observeOn(AndroidSchedulers.mainThread())
}

fun <T> Observable<T>.schedulersTrampolineE(): Observable<T> {
    return this.subscribeOn(Schedulers.trampoline())
        .observeOn(AndroidSchedulers.mainThread())
        .onErrorResumeNext()
}

/**
 * 拦截错误统一处理
 */
fun <T> Observable<T>.onErrorResumeNext(): Observable<T> {
    return this.onErrorResumeNext { t ->
        Observable.error(ExceptionHandle.getInstance().handleException(t))
    }
}

/**
 * 订阅观察者
 */
fun <T> Observable<T>.subscriptions(observer: Observer<in T>) {
    this.onErrorResumeNext()
        .subscribe(observer)
}

/**
 * 订阅
 * @param onNext 消费者
 * @param onError 错误通知
 * @param onComplete  完成通知
 */
@JvmOverloads
fun <T> Observable<T>.subscriptions(
    onNext: ((t: T?) -> Unit)? = null,
    onError: ((throwable: Throwable?) -> Unit)? = null,
    onComplete: (() -> Unit)? = null
): Disposable {
    this.onErrorResumeNext()
    val callback = ExceptionHandle.getInstance().callback
    return this.subscribe({
        if (it is BaseResponse<*>) {
            if (callback?.invoke(it.code, it.msg, onError, it.data) != true) {
                onNext?.invoke(it)
            }
        } else {
            onNext?.invoke(it)
        }
    }, {
        onError?.invoke(it)
    }, {
        onComplete?.invoke()
    })
}

/**
 * 订阅IO线程
 * @param onNext 消费者
 * @param onError 错误通知
 * @param onComplete  完成通知
 */
@JvmOverloads
fun <T> Observable<T>.subscribeIO(
    onNext: ((t: T?) -> Unit)? = null,
    onError: ((throwable: Throwable?) -> Unit)? = null,
    onComplete: (() -> Unit)? = null
): Disposable {
    return this.schedulersIO()
        .subscriptions(onNext, onError, onComplete)
}

fun <T> Observable<T>.subscribeIO(observer: Observer<in T>) {
    this.schedulersIO()
        .subscriptions(observer)
}

/**
 * 订阅Computation线程
 * @param onNext 消费者
 * @param onError 错误通知
 * @param onComplete  完成通知
 */
@JvmOverloads
fun <T> Observable<T>.subscribeComputation(
    onNext: ((t: T?) -> Unit)? = null,
    onError: ((throwable: Throwable?) -> Unit)? = null,
    onComplete: (() -> Unit)? = null
): Disposable {
    return this.schedulersComputation()
        .subscriptions(onNext, onError, onComplete)
}

fun <T> Observable<T>.subscribeComputation(observer: Observer<in T>) {
    this.schedulersComputation()
        .subscriptions(observer)
}

/**
 * 订阅New线程
 * @param onNext 消费者
 * @param onError 错误通知
 * @param onComplete  完成通知
 */
@JvmOverloads
fun <T> Observable<T>.subscribeNew(
    onNext: ((t: T?) -> Unit)? = null,
    onError: ((throwable: Throwable?) -> Unit)? = null,
    onComplete: (() -> Unit)? = null
): Disposable {
    return this.schedulersNewThread()
        .subscriptions(onNext, onError, onComplete)
}

fun <T> Observable<T>.subscribeNew(observer: Observer<in T>) {
    this.schedulersNewThread()
        .subscriptions(observer)
}

/**
 * 订阅Trampoline线程
 * @param onNext 消费者
 * @param onError 错误通知
 * @param onComplete  完成通知
 */
@JvmOverloads
fun <T> Observable<T>.subscribeTrampoline(
    onNext: ((t: T?) -> Unit)? = null,
    onError: ((throwable: Throwable?) -> Unit)? = null,
    onComplete: (() -> Unit)? = null
): Disposable {
    return this.schedulersTrampoline()
        .subscriptions(onNext, onError, onComplete)
}

fun <T> Observable<T>.subscribeTrampoline(observer: Observer<in T>) {
    this.schedulersTrampoline()
        .subscriptions(observer)
}